Add large-payload blob auto-purge (opt-in singleton job, worker/SDK side) by YunchuWang · Pull Request #758 · microsoft/durabletask-dotnet

wangbill (YunchuWang) · 2026-07-08T19:51:11Z

Summary

Large orchestration payloads are externalized to Azure Blob Storage by the AzureBlobPayloads extension, with a token persisted in SQL instead of the payload bytes. When the orchestration is purged, DTS removes the SQL state but cannot delete the backing blob — it has no customer storage credentials. Only the worker does.

This PR implements the worker/SDK side. The backend records a durable tombstone for each externalized payload whose orchestration state is gone; the worker fetches due tombstones, deletes the blobs, and reports the outcome of every row so the backend can resolve, reschedule, or quarantine it.

Companion changes: contract in microsoft/durabletask-protobuf#76, backend in AAPT-DTMB PR 16368738.

gRPC contract

Two unary RPCs on TaskHubSidecarService (worker is the client). The vendored src/Grpc/orchestrator_service.proto is byte-identical to protobuf#76 — verified mechanically by exact-substring comparison, not by eye.

rpc GetLargePayloadTombstones(GetLargePayloadTombstonesRequest) returns (GetLargePayloadTombstonesResponse);
rpc ReportLargePayloadPurgeResults(ReportLargePayloadPurgeResultsRequest) returns (ReportLargePayloadPurgeResultsResponse);

LargePayloadTombstone { partitionId, instanceKey, payloadId, token, revision }
LargePayloadPurgeResult { identity, revision, disposition }
Opt-in is google.protobuf.BoolValue large_payload_auto_purge_enabled = 12 on the existing GetWorkItemsRequest — no new handshake, and null means "no opinion".

revision is echoed back unmodified as a compare-and-swap guard, so duplicate or stale reports are no-ops without a per-row lease.

Dispositions

Three dispositions, split on whether a failure can self-heal. There is no Discarded: a success is reported explicitly as Deleted, and every failure is carried by Retry or Quarantined.

Disposition	Meaning	Examples
`Deleted`	Tombstone resolved	blob deleted; already absent; blob not store-owned
`Retry`	Transient; backend reschedules	storage unreachable, 401/403, 408/429/5xx, unknown token prefix
`Quarantined`	Deterministic; retrying can never succeed	malformed v2 token, HTTP 400, legacy v1 token

The worker never computes a retry delay. It reports the failure and the backend owns scheduling and backoff. The orchestrator reports the whole batch unconditionally — including retryable rows — because the backend needs to hear about a failure in order to defer the row. If every row comes back Retry (a storage outage), the cycle applies ErrorBackoff so an outage cannot become a tight refetch loop.

disposition is the entire outcome. An earlier revision also carried reason and storageErrorCode; both were removed after verifying they were write-only end to end — the backend persists them and nothing reads either one. Failure detail is logged by the worker instead, at the classification site, which is strictly richer than the enum was: 401/403, 5xx, and an unreachable account are distinct in telemetry where the enum collapsed all three into one value.

Blob ownership marker

A recognized token proves only that the text looks like one this store emits — the column is customer-writable. So ownership is recorded on the object itself: UploadAsync writes fixed blob metadata managed_by=dts, and the worker re-reads the target's metadata immediately before deleting.

Marker present → delete, using If-Match on the ETag from that same read, so a mid-flight overwrite fails the delete rather than destroying newer content.
Marker absent → blob left untouched, and the row is still reported Deleted, so the tombstone is resolved rather than retried forever. This is an expected outcome, not a defect, and must not be quarantined. The worker logs it distinctly so a customer whose payloads are all self-authored is still visible in telemetry.

The metadata name uses an underscore because Azure requires blob metadata names to be valid C# identifiers; managed-by would be rejected at upload.

Only blob:v2: tokens are auto-purged. A v1 token reaching this path is an invariant violation (v1 is excluded at insertion) and is quarantined rather than deleted.

Testing

Verified on a clean (--no-incremental) build:

dotnet build Microsoft.DurableTask.sln — 0 errors
test/Extensions/AzureBlobPayloads.Tests — 94 passed
test/Client/Grpc.Tests — 56 passed
Blame-attributed warnings — 127 unique in-repo, exactly 1 introduced by this branch

LargePayloadPurgeEnumParityTests pins proto↔managed parity for LargePayloadPurgeDisposition by value and name in both directions, and asserts that no inbound type exposes an enum. Disposition is the one enum crossing the wire, and its numeric cast in ReportLargePayloadPurgeResultsActivity is what decides whether a row is deleted or quarantined; that cast is safe only because enums travel outbound-only, and the test fails the build if a future change breaks that invariant.

Notes / intentional deviations

Ships one intentional warning: CA1873 at BlobPurgeJobOrchestrator.cs:88 (unguarded logging). Kept deliberately for consistency with 77 existing instances across the solution. Blame-based attribution against main confirms this is the only warning this branch introduces.
PurgedCount counts every Deleted row, including blobs skipped for lacking the ownership marker. Excluding them would pin the counter at 0 for a customer whose payloads are all self-authored, making a healthy draining job read as wedged; the worker log supplies the precision instead.
PayloadStore.DeleteAsync is virtual with a default that throws NotSupportedException, so existing external subclasses are unaffected.
BlobPurgeJob.Create is a no-op when already Active so racing client processes don't disturb a running job.
BlobPurgeJobStarter implements IDisposable rather than disposing its CTS in StopAsync: that method returns on the host shutdown token while the ensure task may still be live, so disposing there would fault it with an unobserved ObjectDisposedException.

Depends on protobuf#76 for the authoritative contract and on the DTS backend serving these RPCs.

Large orchestration payloads are externalized to Azure Blob Storage as `blob:v1:<container>:<blobName>` tokens. The DTS backend stores those tokens but cannot delete the backing blobs (it has no storage credentials) — only this SDK can. This adds an opt-in, whole-scheduler singleton durable entity + orchestration job (mirroring src/ExportHistory) that drains payload rows the backend has soft-deleted and deletes their blobs, then acks so the backend can hard-delete the rows. Design: - PayloadStore.DeleteAsync is virtual (default throws NotSupportedException so it is non-breaking for existing external subclasses); BlobPayloadStore overrides it to decode the token and call DeleteIfExistsAsync (idempotent). - BlobPurgeJob (TaskEntity singleton): Create is a no-op when already Active so racing client processes don't disturb the running job; Run starts a fixed-id orchestrator. - BlobPurgeJobOrchestrator (perpetual): fetch a batch of tombstones, delete the blobs with capped parallelism, ack the successful deletions (failed tokens stay tombstoned to retry), idle on a timer when empty, ContinueAsNew periodically. - ExecuteBlobPurgeJobOperationOrchestrator bridges client -> entity. - Two new unary RPCs on TaskHubSidecarService: GetTombstonedPayloads / AckPurgedPayloads (authoritative proto follow-up: microsoft/durabletask-protobuf#76). - LargePayloadStorageOptions gains AutoPurge (opt-in, default false) and PayloadPurgeBatchSize (default 500). - Client-side BlobPurgeJobStarter (IHostedService) ensures the singleton job when AutoPurge is enabled, without blocking host startup. Worker always registers the entity/orchestrators/activities so a client-enabled job has something to run. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

…er simplification - Drop the `Dto` suffix now that the payload records are first-class public types in `Microsoft.DurableTask.Client` (`TombstonedPayload`, `PayloadPurgeAck`). - Collapse the magic `500` batch-size literal into a single `BlobPurgeConstants.DefaultBatchSize` used everywhere. - Rename `BlobPurgeJobStatus.Stopped` -> `Pending` (still the zero value) and remove the dead `Failed` member (nothing ever set it; the job self-heals). - Make the perpetual orchestrator self-heal: wrap each cycle in try/catch so a transient backend/entity/activity failure logs, backs off, and continues instead of failing the orchestration and killing the eternal loop. - Ack poison tokens: `DeleteExternalBlobActivity` now returns a three-way `BlobDeleteResult` (Deleted/Discarded/Retry). Malformed tokens are discarded and acked so the backend can clear the stuck row instead of re-streaming it forever; transient failures stay tombstoned to retry. - Replace the single-value `BlobPurgeJobCreationOptions` record with a plain `int` on `BlobPurgeJob.Create`. - Guard the client fetch RPC: `GetTombstonedPayloadsAsync` throws `ArgumentOutOfRangeException` unless `0 < limit < 1000`. - Simplify `BlobPurgeJobStarter` to a fixed-instance-id fire-once: drop the entity-active pre-check and schedule the Create bridge once with a fixed instance id, retrying only until the backend is reachable. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

The two auto-purge activities injected a concrete DurableTaskClient, but AddDurableTaskWorker never registers one - only AddDurableTaskClient does. In a worker-only ("client triggers, worker executes") split deployment with auto-purge enabled, resolving either activity threw at dispatch time - and only at dispatch, because DurableTaskRegistry stores a lazy ActivatorUtilities.GetServiceOrCreateInstance factory - so auto-purge was silently 100% broken and blobs accumulated forever. Introduce an internal ILargePayloadPurgeClient over exactly the two backend RPCs the job needs (fetch tombstones, report results) plus a gRPC implementation that reuses the worker's OWN transport: it builds a TaskHubSidecarServiceClient from the named GrpcDurableTaskWorkerOptions, mirroring the worker's Channel -> CallInvoker precedence, and resolves it lazily on first RPC so the activities stay constructible in hosts that never run the job. The marshalling (limit 1..1000 validation, empty short-circuit, RpcException/Cancelled -> OperationCanceledException, by-value disposition mapping) is copied verbatim from GrpcDurableTaskClient so the wire behavior is identical; its public virtuals and overrides are left untouched. Both purge activities now inject ILargePayloadPurgeClient instead of DurableTaskClient and become internal sealed (no new public API surface; the test project already sees internals via the auto-generated InternalsVisibleTo). Registers the client in the worker's UseExternalizedPayloadsCore against builder.Name. Adds a worker-only regression test asserting both activities construct via the same reflection path the registry uses at dispatch. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated no new comments.

Suppressed comments (4)

src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs:87

The interceptor wiring throws unless GrpcDurableTaskClientOptions.Channel or .CallInvoker is pre-populated. However, the standard UseGrpc(string address) path sets only GrpcDurableTaskClientOptions.Address, which means enabling UseExternalizedPayloads can fail later during options materialization with this ArgumentException even though the client itself supports Address-only configuration.

        // Wrap the gRPC CallInvoker with our interceptor when using the gRPC client
        builder.Services
            .AddOptions<GrpcDurableTaskClientOptions>(builder.Name)
            .PostConfigure<PayloadStore, IOptionsMonitor<LargePayloadStorageOptions>>((opt, store, monitor) =>
            {
                LargePayloadStorageOptions opts = monitor.Get(builder.Name);
                if (opt.Channel is not null)
                {
                    Grpc.Core.CallInvoker invoker = opt.Channel.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts));
                    opt.CallInvoker = invoker;

                    // Ensure client uses the intercepted invoker path
                    opt.Channel = null;
                }
                else if (opt.CallInvoker is not null)
                {
                    opt.CallInvoker = opt.CallInvoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts));
                }
                else
                {
                    throw new ArgumentException(
                        "Channel or CallInvoker must be provided to use Azure Blob Payload Externalization feature");
                }

src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskWorkerBuilderExtensions.AzureBlobPayloads.cs:85

The worker-side interceptor wiring throws unless GrpcDurableTaskWorkerOptions.Channel or .CallInvoker is set. But the common builder API UseGrpc(string address) configures only GrpcDurableTaskWorkerOptions.Address, so UseExternalizedPayloads can break otherwise-valid Address-only worker configurations when options are resolved.

        // Wrap the gRPC CallInvoker with our interceptor when using the gRPC worker
        builder.Services
            .AddOptions<GrpcDurableTaskWorkerOptions>(builder.Name)
            .PostConfigure<PayloadStore, IOptionsMonitor<LargePayloadStorageOptions>>((opt, store, monitor) =>
            {
                LargePayloadStorageOptions opts = monitor.Get(builder.Name);
                if (opt.Channel is not null)
                {
                    var invoker = opt.Channel.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts));
                    opt.CallInvoker = invoker;

                    // Ensure worker uses the intercepted invoker path
                    opt.Channel = null;
                }
                else if (opt.CallInvoker is not null)
                {
                    opt.CallInvoker = opt.CallInvoker.Intercept(new AzureBlobPayloadsSideCarInterceptor(store, opts));
                }
                else
                {
                    throw new ArgumentException(
                        "Channel or CallInvoker must be provided to use Azure Blob Payload Externalization feature");
                }

src/Client/Grpc/GrpcDurableTaskClient.cs:685

ReportLargePayloadPurgeResultsAsync forwards Disposition by numeric cast with no validation. The proto contract explicitly treats 0/UNSPECIFIED as invalid and says the backend must reject it; without a guard, a default-initialized LargePayloadPurgeResult (or a consumer mistake) will fail with a backend error rather than a clear client-side exception.

        P.ReportLargePayloadPurgeResultsRequest request = new();
        foreach (LargePayloadPurgeResult result in results)
        {
            request.Results.Add(new P.LargePayloadPurgeResult
            {
                PartitionId = result.PartitionId,
                InstanceKey = result.InstanceKey,
                PayloadId = result.PayloadId,
                Revision = result.Revision,

                // The managed disposition enum declares the same numeric values as its protobuf counterpart,
                // so it maps across by value. This is the only enum on the message and it only travels
                // outbound, so the SDK can never receive a value it does not know.
                Disposition = (P.LargePayloadPurgeDisposition)result.Disposition,
            });

src/Extensions/AzureBlobPayloads/AutoPurge/Client/GrpcLargePayloadPurgeClient.cs:98

Same as GrpcDurableTaskClient: this request builder casts Disposition with no validation. Since the backend rejects UNSPECIFIED/0, adding a client-side guard would prevent hard-to-diagnose backend failures if a default/invalid LargePayloadPurgeResult ever reaches this internal transport.

        P.ReportLargePayloadPurgeResultsRequest request = new();
        foreach (LargePayloadPurgeResult result in results)
        {
            request.Results.Add(new P.LargePayloadPurgeResult
            {
                PartitionId = result.PartitionId,
                InstanceKey = result.InstanceKey,
                PayloadId = result.PayloadId,
                Revision = result.Revision,

                // The managed disposition enum declares the same numeric values as its protobuf counterpart,
                // so it maps across by value. This is the only enum on the message and it only travels
                // outbound, so the SDK can never receive a value it does not know.
                Disposition = (P.LargePayloadPurgeDisposition)result.Disposition,
            });

ILargePayloadPurgeClient had exactly one implementation and no mocks, and GrpcLargePayloadPurgeClient's lazy GetClient()/lock/cache machinery merely duplicated what DI already does: a TryAddSingleton factory is already lazy (it runs on first resolve, which is at dispatch) and DI already caches the singleton. Register the worker's TaskHubSidecarServiceClient directly from the named GrpcDurableTaskWorkerOptions (CallInvoker, then Channel) and inline the tombstone/report marshalling into the two activities, which now inject the sidecar client instead of the removed abstraction. Behavior is unchanged: the RPCs still ride the worker's own transport, so a worker-only host runs the purge job with no DurableTaskClient present. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 36 out of 36 changed files in this pull request and generated no new comments.

Suppressed comments (1)

src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs:303

SignalJobStopAsync unconditionally dereferences client.Entities. If DurableTaskClientOptions.EnableEntitySupport is false (e.g., overridden after calling UseExternalizedPayloads), GrpcDurableTaskClient.Entities throws NotSupportedException, and this method will fall into the outer retry loop forever without ever issuing the stop request. Since stopping a previously-started purge job is correctness-critical, handle the "entities disabled" case explicitly and still request a stop (e.g., via the bridge orchestration).

                // Held in a local so the query and the signal below are issued against the same object. Both go
                // through this one property, which is also where every "entities are not supported" gate in this
                // SDK lives, so a client that cannot answer the query cannot receive the signal either.
                DurableEntityClient entities = client.Entities;

During a mixed rollout an older DTS backend (or a stale local emulator image) returns gRPC Unimplemented for the new large-payload purge RPCs. The fetch/report activities previously translated only Cancelled, so the purge job retried the unsupported operation forever (~1m45s loop) with no actionable diagnostic. Map Unimplemented to NotImplementedException and genuinely stop the job via a new terminal Unsupported state. - BlobPurgeJobStatus: add terminal Unsupported, distinct from Pending so Create refuses to re-activate it. - BlobPurgeJob: Create guards against Unsupported (the reconciler re-issues Create every interval, so this guard is what makes the stop real); add idempotent MarkUnsupported and guarded Reset ops. - Both purge activities and both GrpcDurableTaskClient overrides map StatusCode.Unimplemented -> NotImplementedException (client overrides follow the existing bare-Detail precedent). - Orchestrator: retry policy HandleFailure skips NotImplementedException; a dedicated catch disables the job via awaited MarkUnsupported and exits the loop cleanly with a dedicated diagnostic. - BlobPurgeJobStarter: signal Reset once per process start so an upgraded backend recovers on restart (benign, self-correcting race documented). - Logs: EventIds 827-830. Tests: entity Create-does-not-resurrect (failing-first), MarkUnsupported idempotency, Reset clears/no-ops on Active; activity + client Unimplemented mapping; orchestrator disables-and-exits. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated no new comments.

Suppressed comments (1)

src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs:163

StopAsync waits for either the background task or the host shutdown token, but it never awaits the background task when it completes. If the background task faults (even after cancellation), the exception can remain unobserved and may surface later via UnobservedTaskException. Consider observing the task when it wins the race, while still swallowing exceptions as intended.

        Task? pending = this.backgroundTask;
        if (pending is not null)
        {
            // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled
            // result.
            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);
        }

…ck time Addresses two performance findings in the large-payload blob auto-purge job. Thread 1 (per-blob activity fan-out): the orchestrator previously issued one CallActivityAsync per tombstone. A 1000-row batch meant 1000 activity calls (~2000 history events) per cycle. DeleteExternalBlobActivity now accepts a chunk of tokens (List<string>) and returns one outcome per token (List<BlobPurgeOutcome>), and the orchestrator fans out in chunks of DeleteChunkSize=50 -> 20 activity calls (~40 history events), a 50x reduction. Total concurrent storage deletes are held at today's 32: MaxParallelChunkActivities=4 (orchestrator) x MaxParallelDeletesPerChunk=8 (inside each chunk activity) = 32. Results stay per-token: the chunk activity writes each outcome into a pre-sized array by input index, and the orchestrator asserts the returned count equals the sent count before zipping so a mismatch fails loudly instead of misattributing dispositions. One failing token in a chunk never fails its peers (every branch still returns a disposition; the activity never throws). Chunk-level retry stays safe because deletion is idempotent. The per-token classification body is moved verbatim; this is a fan-out change, not a semantics change. The Azure Blob Batch suggestion is intentionally not implemented (extra package + bulk store API + per-sub-request status semantics; the chunking already removes the orchestration-history problem). Thread 2 (unbounded per-delete retry budget): a single hung blob delete could burn up to ~18 minutes (9 attempts x 2 min) while holding a parallelism slot, because the store delete was called with CancellationToken.None. Each delete is now bounded by a CancellationTokenSource with SingleDeleteTimeout=60s. A timeout surfaces as TaskCanceledException, which the existing catch-all maps to a Retry disposition, so a persistently slow blob backs off server-side instead of stalling the wave. TaskActivityContext exposes no ambient CancellationToken, so the timeout source stands alone (host shutdown still tears the activity down). PUBLIC API BREAK (unreleased feature): DeleteExternalBlobActivity changes from TaskActivity<string, BlobPurgeOutcome> to TaskActivity<List<string>, List<BlobPurgeOutcome>>. No public-API baseline file exists in the repo, so nothing to update. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated no new comments.

Suppressed comments (3)

src/Client/Grpc/GrpcDurableTaskClient.cs:689

ReportLargePayloadPurgeResultsAsync currently forwards LargePayloadPurgeDisposition values directly onto the wire by numeric cast, including Unspecified (0). The proto contract explicitly treats 0/UNSPECIFIED as invalid (backend must reject it), so it would be better to fail fast with a clear ArgumentException before issuing the RPC.

        foreach (LargePayloadPurgeResult result in results)
        {
            request.Results.Add(new P.LargePayloadPurgeResult
            {
                PartitionId = result.PartitionId,
                InstanceKey = result.InstanceKey,
                PayloadId = result.PayloadId,
                Revision = result.Revision,

                // The managed disposition enum declares the same numeric values as its protobuf counterpart,
                // so it maps across by value. This is the only enum on the message and it only travels
                // outbound, so the SDK can never receive a value it does not know.
                Disposition = (P.LargePayloadPurgeDisposition)result.Disposition,
            });
        }

src/Extensions/AzureBlobPayloads/AutoPurge/Activities/ReportLargePayloadPurgeResultsActivity.cs:53

ReportLargePayloadPurgeResultsActivity forwards dispositions by numeric cast, but does not guard against LargePayloadPurgeDisposition.Unspecified (0). Since the backend contract rejects UNSPECIFIED, validate and throw before issuing the RPC so this fails fast with a clear message and doesn’t produce an avoidable backend error.

        P.ReportLargePayloadPurgeResultsRequest request = new();
        foreach (LargePayloadPurgeResult result in input)
        {
            request.Results.Add(new P.LargePayloadPurgeResult
            {
                PartitionId = result.PartitionId,
                InstanceKey = result.InstanceKey,
                PayloadId = result.PayloadId,
                Revision = result.Revision,

                // The managed disposition enum declares the same numeric values as its protobuf counterpart,
                // so it maps across by value. This is the only enum on the message and it only travels
                // outbound, so the SDK can never receive a value it does not know.
                Disposition = (P.LargePayloadPurgeDisposition)result.Disposition,
            });
        }

src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs:54

DeleteExternalBlobActivity is declared public, unlike the other auto-purge activities in this feature (which are internal). Since this type is new and appears to be an implementation detail registered via UseExternalizedPayloads, consider making it internal sealed to avoid expanding the public surface area unnecessarily.

[DurableTask]
public class DeleteExternalBlobActivity(
    PayloadStore store,
    ILogger<DeleteExternalBlobActivity> logger)
    : TaskActivity<List<string>, List<BlobPurgeOutcome>>

…nstant, test null-forgiving Three small, independent review items; item 1 is documentation-only with zero behavior change. Item 1 (r3677836542 - blob versioning / soft delete): IncludeSnapshots deletes snapshots and the base blob but not blob *versions*, and soft delete retains bytes container-side regardless of how the delete is issued, so acknowledging a purge does not guarantee reclamation. Documented this rather than enumerating versions, because even deleting every version would not guarantee reclamation under a soft-delete/retention policy - immediate reclamation is unobtainable client-side and belongs to a lifecycle-management policy. Changes are doc-only: - PayloadDeleteOutcome.Deleted now states it means the delete was accepted by storage, not that bytes were reclaimed. - LargePayloadStorageOptions.AutoPurge XML doc gains a paragraph on versioning/soft delete and recommends a lifecycle policy. - An inline comment at BlobPayloadStore's DeleteIfExistsAsync call records the IncludeSnapshots-vs-versions distinction and why versions are not enumerated. Item 2 (r3767474415 - move the 1000 limit to a shared constant): the GetLargePayloadTombstones request limit was hard-coded as the literal 1000 in two validators, each with a duplicated message string, so the client and the auto-purge extension could silently drift. Introduced a single public authority LargePayloadTombstone.MaxRequestLimit = 1000 (in Client core, reachable by both call sites) and pointed both validators at it, including building the exception message from it. The extension's internal BlobPurgeConstants.MaxBatchSize now delegates to the same constant instead of redefining 1000, so there is one source of truth. Adds public API surface (the new public const); the repo has no public-API baseline file to update. Item 3 (r3772204948 - missing null-forgiving operator): BlobPurgeJobStarterTests line 82 accessed run.Signal.Value without the '!' that line 81 uses; added it for consistency. One character; no behavior change. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated 1 comment.

…urge paths Close the remaining auto-purge test gaps with tests only; no product code changes. - Named client and worker builders: resolve LargePayloadStorageOptions, the entity-support flip, the intercepted gRPC options, the starter/store and the purge activities under a non-empty builder name, plus a negative test proving configuration does not leak to other names or the default name. - Combined client + worker host: exactly one shared PayloadStore and one client-registered BlobPurgeJobStarter (not one per builder), with the purge activity still constructible. - Empty fetch: an Active job whose fetch returns nothing idles on the timer without invoking the delete or report activities. - Continue-as-new bound: the perpetual orchestrator runs exactly ContinueAsNewFrequency cycles before a single ContinueAsNew that resets the cycle count and carries the job identity and batch size forward. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated no new comments.

Suppressed comments (1)

src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs:268

The exception message here says the payload lives in a different storage account, but this branch is taken whenever the token’s container URI doesn’t exactly match the configured container URI (which can also be the same account with a different container). This can mislead operators when diagnosing why deletes are being rejected.

            throw new PayloadStorageException(
                $"The externalized payload lives in a different storage account ('{decoded.ContainerUri}') than the " +
                $"currently-configured payload store ('{this.containerClient.Uri}'). Cross-account payload deletes " +
                "require identity (AAD) authentication with access to both accounts; connection-string / " +
                "account-key credentials are account-specific and cannot delete in another account.");

Blob soft delete and blob versioning are storage-account-level settings (Blob service properties), not container-level. Correct the three doc/comment sites added for the auto-purge storage-semantics note so customers reading the public XML docs are not misinformed: - LargePayloadStorageOptions.AutoPurge <remarks> - PayloadDeleteOutcome.Deleted <summary> - BlobPayloadStore delete comment above DeleteIfExistsAsync The correction strengthens the existing argument: because these are account-level policies, a single container cannot opt out, so retained versions / soft-deleted bytes are even further outside the SDK's control. Also clarify that reclamation is governed by an account lifecycle-management policy whose rule can be scoped to the payload container's blob prefix. Comments / XML docs only; no code, signature, or behavior change. IncludeSnapshots is unchanged and version enumeration remains deliberately not implemented. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

The virtual PayloadStore.DeleteAsync <remarks> already binds third-party stores to idempotency and ownership (NotStoreOwned), but its <returns> says only "whether the object was deleted" - silent on what a successful delete actually guarantees, which is exactly the ambiguity flagged in review. Add product-neutral contract prose to the <remarks>: a Deleted outcome means the store no longer references the object and the underlying storage accepted the delete, NOT that the bytes were reclaimed; storage-level retention features such as versioning, soft delete, or retention policies may keep the content for a policy-defined period, and implementations are not expected to defeat them. Store-agnostic by design - PayloadStore is a public extensibility point - with no Azure/blob/container specifics, mirroring the product-neutral NotStoreOwned naming. The Azure-specific reclamation semantics remain documented on the concrete BlobPayloadStore path (inline comment, PayloadDeleteOutcome.Deleted, AutoPurge remarks), so this adds the missing base-contract sentence without duplicating them. Docs only; no code, signature, or behavior change. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated no new comments.

Suppressed comments (2)

src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs:163

StopAsync waits for the background reconciliation task to finish but never observes its result. If the background task faults (e.g., unexpected exception escapes the loop), the exception remains unobserved and can surface later via TaskScheduler.UnobservedTaskException. It’s safer to observe/ignore the completion explicitly once the task has finished.

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        this.cts?.Cancel();

        Task? pending = this.backgroundTask;
        if (pending is not null)
        {
            // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled
            // result.
            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);
        }

src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs:268

This exception message claims the token points to a “different storage account”, but the comparison is against the full container URI (scheme/host/port/path). A token in the same account but a different container would also hit this branch, making the message/action guidance misleading. Consider wording this as “different storage account or container”, or splitting account-vs-container mismatch if you want to keep the stronger wording.

            throw new PayloadStorageException(
                $"The externalized payload lives in a different storage account ('{decoded.ContainerUri}') than the " +
                $"currently-configured payload store ('{this.containerClient.Uri}'). Cross-account payload deletes " +
                "require identity (AAD) authentication with access to both accounts; connection-string / " +
                "account-key credentials are account-specific and cannot delete in another account.");

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated 3 comments.

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated no new comments.

Suppressed comments (1)

src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/ExecuteBlobPurgeJobOperationOrchestrator.cs:22

RunAsync calls CallEntityAsync(...) and returns it as object. With the default JsonDataConverter, deserializing JSON into typeof(object) produces JsonElement/primitive wrappers rather than the entity operation's actual return type, so this orchestrator does not reliably “return the result” as its XML doc claims. Since the only current caller (BlobPurgeJobStarter) uses this orchestrator for side-effects only, prefer the non-generic CallEntityAsync overload and return null to avoid type ambiguity and unnecessary deserialization.
```
    public override async Task<object> RunAsync(
        TaskOrchestrationContext context, BlobPurgeJobOperationRequest input)
    {
        return await context.Entities.CallEntityAsync<object>(input.EntityId, input.OperationName, input.Input);
    }
```

…yloads Externalized payloads are configured per host, not per named builder: the PayloadStore and the purge TaskHubSidecarServiceClient are registered as container-wide singletons, so the first builder that calls UseExternalizedPayloads supplies the configuration the whole host shares. Configuring multiple named workers/clients in one host with different storage accounts or different backends is therefore not supported - later builders silently reuse the first registration. Adds class-level <remarks> to both the worker and client DurableTask*BuilderExtensionsAzureBlobPayloads classes stating the constraint, and extends the inline comment above the worker's sidecar-client TryAddSingleton to explain why a keyed/named registration cannot work here: the consuming purge activities are constructed from the plain IServiceProvider with no worker name in scope, so a keyed client would have no resolvable consumer. Comments and XML docs only - no behavior, options, or DI registration change. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 38 out of 38 changed files in this pull request and generated no new comments.

Suppressed comments (2)

src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs:129

RegisterBlobPurgeJobStarter adds a new IHostedService every time UseExternalizedPayloads() is called. Because PayloadStore is registered via TryAddSingleton (first builder wins), multiple named client builders in the same host can end up with multiple BlobPurgeJobStarter instances that each read different named options (and may try to start vs stop the same singleton job), while all sharing the same PayloadStore. This contradicts the per-host singleton behavior described in the remarks and can lead to conflicting reconciliation behavior.

    static void RegisterBlobPurgeJobStarter(IDurableTaskClientBuilder builder)
    {
        string builderName = builder.Name;
        builder.Services.AddSingleton<IHostedService>(sp => new BlobPurgeJobStarter(
            sp.GetRequiredService<IDurableTaskClientProvider>(),
            sp.GetRequiredService<PayloadStore>(),
            sp.GetRequiredService<IOptionsMonitor<LargePayloadStorageOptions>>(),
            builderName,
            sp.GetRequiredService<ILogger<BlobPurgeJobStarter>>()));
    }

src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs:164

StopAsync claims it will "swallow any faulted/cancelled result", but it never observes backgroundTask exceptions (it only awaits Task.WhenAny). If the background task faults before/while StopAsync runs, the exception can remain unobserved and surface later as an UnobservedTaskException. Consider awaiting the task when it completes (and catching) so faults are actually observed.

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        this.cts?.Cancel();

        Task? pending = this.backgroundTask;
        if (pending is not null)
        {
            // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled
            // result.
            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);
        }

…ent-facing purge APIs F1: DeleteExternalBlobActivity now quarantines a blob that carries no ownership marker (PayloadDeleteOutcome.NotStoreOwned) instead of reporting it Deleted. The ownership marker is new in this PR while main has been emitting v2 tokens without writing it, so on upgrade every pre-existing v2 blob reaches this path; reporting Deleted would make the backend hard-delete the tombstone row, orphaning the blob permanently and destroying the only durable record of it. Quarantine preserves the row and its token backend-side as evidence, matching the v1 and malformed-token branches. Removes the now-unused BlobPurgeBlobNotStoreOwned log (EventId 821). F2: BlobPurgeJobStarter.SignalJobStopAsync now reconciles on a fixed interval like the enabled path instead of signalling Stop once and returning. AutoPurge is per-host config, so during a rolling deployment a replica still on the enabled path re-creates the job every interval; a disabled replica that stopped it once would let it come straight back and the fleet would never converge. The loop is governed by the same cancellationToken as the enabled path, so shutdown ends it, and each pass reads state first so a job that is already stopped is a no-op read. F3: removes the client-facing GetLargePayloadTombstonesAsync and ReportLargePayloadPurgeResultsAsync from DurableTaskClient and their GrpcDurableTaskClient overrides. The purge activities inject the protobuf TaskHubSidecarServiceClient directly and carry their own enum mapping, so these virtuals were a duplicate of the product path, not the product path. The LargePayloadTombstone, LargePayloadPurgeResult, and LargePayloadPurgeDisposition contract types are kept. Deletes the duplicate LargePayloadPurgeUnimplementedTests (the same Unimplemented to NotImplementedException behavior is covered on the actual activity path by PurgeActivityUnimplementedTests) and repoints the enum-parity test at ReportLargePayloadPurgeResultsActivity. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated no new comments.

Suppressed comments (2)

src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs:167

StopAsync awaits Task.WhenAny(...) but never observes exceptions from the background task. If the reconcile loop faults, this can surface as an unobserved task exception (process-level event / potential crash depending on runtime settings). Swallowing faults is fine here, but the exception should still be observed when the task completes.

        if (pending is not null)
        {
            // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled
            // result.
            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);

src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs:182

PayloadDeleteOutcome.NotStoreOwned is mapped to LargePayloadPurgeDisposition.Quarantined, but the contract semantics (proto comment for LARGE_PAYLOAD_PURGE_DISPOSITION_DELETED and managed LargePayloadPurgeDisposition.Deleted docs) treat "blob not store-owned" as a terminal success that should be reported as Deleted (tombstone resolved). Quarantining this case will unnecessarily move rows into the operator-only path and diverges from the documented wire meaning. (Any tests asserting Quarantined for NotStoreOwned should be updated accordingly.)

            if (outcome == PayloadDeleteOutcome.NotStoreOwned)
            {
                this.logger.BlobPurgeDeleteQuarantined("NotStoreOwned", null);
                return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Quarantined);
            }

…loop RunDisabledStarterAsync's shutdown comment justified determinism by the background task running to completion on its own. After the disable path was made to reconcile on an interval, the loop no longer self-completes: it parks in the reconcile delay and StopAsync's cancellation is what ends it, after pass one's signal decision. Reword the comment to match; no behavior change. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated 1 comment.

The F1 change (a814ffc) made an unowned/unmarked blob report Quarantined instead of Deleted, but several comments and XML docs still described the old Deleted-covers-unowned behavior. Doc/comment only - no product logic or test-logic change. - src/Grpc/orchestrator_service.proto: DELETED now covers two cases (blob deleted, blob already absent), not three; the "left in place because unowned" case is QUARANTINED now. Language-neutral wire semantics only; the same hunk mirrors into microsoft/durabletask-protobuf #76. - src/Client/Core/LargePayloadPurgeDisposition.cs: same two-case fix on the C# mirror of that enum ("all three cases" -> "both cases"). - src/Extensions/AzureBlobPayloads/PayloadStore/PayloadDeleteOutcome.cs: the NotStoreOwned summary no longer claims the reference "is still resolved"; it states the store-layer fact plus a neutral note that the outcome is not proof of deletion and the caller decides how to dispose of the reference. No disposition named - that mapping is the activity's decision. - src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs: same correction at the HasOwnershipMarker return site. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated no new comments.

Suppressed comments (3)

src/Extensions/AzureBlobPayloads/AutoPurge/Activities/DeleteExternalBlobActivity.cs:182

This maps PayloadDeleteOutcome.NotStoreOwned (missing ownership marker) to Quarantined. However the PR description states that when the marker is absent the blob should be left untouched and the row should still be reported Deleted so the tombstone is resolved (i.e., “must not be quarantined”). Either the implementation or the PR description needs to be updated, because this changes backend behavior (quarantine vs resolve) and alerting semantics.

            // The blob exists but this store never wrote it - it carries no ownership marker - so it is
            // quarantined, not reported as deleted. The marker is new in this PR, but main has been emitting v2
            // tokens without writing it, so on upgrade every pre-existing v2 blob reaches this branch. Reporting
            // Deleted here would tell the backend to hard-delete the tombstone row, orphaning that blob
            // permanently AND destroying the only durable record that it exists. Quarantine keeps the row and its
            // token backend-side as evidence and stops the backend polling it - the same disposition, and for the
            // same reason, as the v1 and malformed-token branches: a delete that cannot be verified as this
            // store's own must not be discarded as a success.
            if (outcome == PayloadDeleteOutcome.NotStoreOwned)
            {
                this.logger.BlobPurgeDeleteQuarantined("NotStoreOwned", null);
                return new BlobPurgeOutcome(LargePayloadPurgeDisposition.Quarantined);
            }

src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs:168

StopAsync uses Task.WhenAny without ever awaiting/observing the background task when it completes. If the background loop faults, the exception can go unobserved (and may surface via UnobservedTaskException) even though the comment says faulted results are swallowed. Consider explicitly awaiting the task when it wins the WhenAny (inside a try/catch) so faults are observed.

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        this.cts?.Cancel();

        Task? pending = this.backgroundTask;
        if (pending is not null)
        {
            // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled
            // result.
            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);
        }
    }

src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs:183

Dispose unconditionally disposes the CancellationTokenSource even if StopAsync returned early due to the host shutdown token, which can still leave the background task running and potentially fault it with ObjectDisposedException (the same risk called out in the remarks for StopAsync). Consider only disposing the CTS once the background task has completed (or avoid disposing it entirely in the timeout case).

    /// Deliberately not disposed in <see cref="StopAsync"/>: that method stops waiting as soon as the host's
    /// shutdown token fires, so the background task may still hold this source's token. Disposing it there would
    /// fault that still-running task with an <see cref="ObjectDisposedException"/> when it next registers a
    /// callback. The container disposes singletons after every <see cref="StopAsync"/> has returned, which is
    /// the safe point.
    /// </remarks>
    public void Dispose()
    {
        this.cts?.Dispose();
    }

The user overruled F1. The backend treats a non-empty quarantine set as a poison/investigate signal: it emits an OTel-flagged LargePayloadPurgeQuarantineOutstanding on every GetLargePayloadTombstones poll (once per IdleDelay = 1 minute, forever), quarantine never auto-expires, and there is no operator tool to clear it. Because the managed_by=dts ownership marker is new in this PR, every pre-existing v2 blob would hit NotStoreOwned on upgrade and permanently wedge that alarm. So an unowned/unmarked blob reports Deleted (terminal success) again, as before F1. Reverts F1 only. F2 (the disable path keeps reconciling) and F3 (the two client-facing purge APIs stay removed) are unchanged. - DeleteExternalBlobActivity.cs, Logs.cs, DeleteExternalBlobActivityTests.cs: restored byte-identical to their pre-F1 (0034bb0) state. NotStoreOwned logs BlobPurgeBlobNotStoreOwned (EventId 821 restored in its 820-822 slot) and falls through to Deleted; the test asserts Deleted again. - orchestrator_service.proto, LargePayloadPurgeDisposition.cs, PayloadDeleteOutcome.cs, BlobPayloadStore.cs: the ce3b547 doc sweep only existed to describe the quarantine semantics, so it is reverted too - restored byte-identical to 9be9931, back to the three-case DELETED wording that matches Deleted semantics and the untouched protobuf #76 copy. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6373d1ba-104b-492a-9737-456a26c8ea34

Copilot

Pull request overview

Copilot reviewed 35 out of 35 changed files in this pull request and generated no new comments.

Suppressed comments (2)

src/Extensions/AzureBlobPayloads/AutoPurge/Client/BlobPurgeJobStarter.cs:167

StopAsync waits for the background task to complete but never observes its final status. If the background task faults, the exception can remain unobserved (potentially surfacing later via UnobservedTaskException). If the task completed before the shutdown token fires, explicitly await it in a try/catch to observe and swallow the exception as intended.

            // The background loop observes cancellation and returns promptly; swallow any faulted/cancelled
            // result.
            await Task.WhenAny(pending, Task.Delay(Timeout.Infinite, cancellationToken)).ConfigureAwait(false);
        }

src/Extensions/AzureBlobPayloads/DependencyInjection/DurableTaskClientBuilderExtensions.AzureBlobPayloads.cs:128

RegisterBlobPurgeJobStarter uses AddSingleton, so calling UseExternalizedPayloads on multiple client builders will register multiple BlobPurgeJobStarter instances. That can start duplicate reconciliation loops and contradicts the per-host singleton behavior described in the remarks. Use a TryAdd-style registration (e.g., TryAddEnumerable) so only one starter is registered per container.

        builder.Services.AddSingleton<IHostedService>(sp => new BlobPurgeJobStarter(

wangbill (YunchuWang) force-pushed the yunchuwang-wangbill-blob-payload-autopurge-sdk branch 2 times, most recently from 82ae04d to 0ac2dc3 Compare July 8, 2026 20:50

github-code-quality Bot found potential problems Jul 8, 2026

View reviewed changes

Comment thread src/Extensions/AzureBlobPayloads/AutoPurge/BlobPayloadAutoPurgeService.cs Fixed

wangbill (YunchuWang) force-pushed the yunchuwang-wangbill-blob-payload-autopurge-sdk branch 4 times, most recently from 0752610 to cab0e9a Compare July 13, 2026 19:07

wangbill (YunchuWang) changed the title ~~Add large-payload blob auto-purge (worker/SDK side)~~ Add large-payload blob auto-purge (opt-in singleton job, worker/SDK side) Jul 13, 2026

github-code-quality Bot found potential problems Jul 13, 2026

View reviewed changes

Comment thread src/Extensions/AzureBlobPayloads/AutoPurge/Orchestrations/BlobPurgeJobOrchestrator.cs Fixed

wangbill (YunchuWang) force-pushed the yunchuwang-wangbill-blob-payload-autopurge-sdk branch from cab0e9a to c05b15a Compare July 13, 2026 20:38

wangbill (YunchuWang) force-pushed the yunchuwang-wangbill-blob-payload-autopurge-sdk branch from c05b15a to 306d19f Compare July 13, 2026 22:43